package com.gzyj.flink.source.mysql;

import com.gzyj.flink.driver.Driverlogininfo;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Date;

/**
 * @author lhj
 */
public class DriverLoginInfoSource extends RichSourceFunction<Driverlogininfo> {

    private PreparedStatement ps;
    private Connection connection;
    private boolean stop = false;


    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);

        try {
            Class.forName("com.mysql.cj.jdbc.Driver");
            ExecutionConfig.GlobalJobParameters globalJobParam = getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
            Configuration globconf = (Configuration) globalJobParam;
            this.connection = DriverManager.getConnection(globconf.getString("mysql-url", null),
                    globconf.getString("mysql-user", "root"),
                    globconf.getString("mysql-password", "123456"));

            String sql = "SELECT\n" +
                    "\tt.* \n" +
                    "FROM\n" +
                    "\t( SELECT VehId, max( LoginTime ) AS LoginTime FROM driverlogininfo GROUP BY VehId ) AS a\n" +
                    "\tLEFT JOIN driverlogininfo AS t ON a.VehId = t.VehId \n" +
                    "\tAND a.LoginTime = t.LoginTime \n" +
                    "ORDER BY\n" +
                    "\tt.LoginTime DESC";


            this.ps = this.connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            this.ps.setFetchSize(50 * 1000);
            this.ps.setMaxRows(50 * 1000);
            this.ps.setFetchDirection(ResultSet.FETCH_REVERSE);
        } catch (Exception e) {
            System.out.println("-----------mysql get connection has exception , msg = " + e.getMessage());
        }
    }

    @Override
    public void run(SourceContext<Driverlogininfo> sourceContext) throws Exception {
        while (!this.stop) {

            ResultSet rs = this.ps.executeQuery();

            while (rs.next()) {
                Driverlogininfo driverlogininfo = new Driverlogininfo();
                driverlogininfo.setCheckObject(rs.getInt("CheckObject"));
                driverlogininfo.setLoginResults(rs.getInt("LoginResults"));
                driverlogininfo.setDriverName(rs.getString("DriverName"));
                driverlogininfo.setDriverNo(rs.getString("DriverNo"));
                driverlogininfo.setLoginTime(rs.getDate("LoginTime") != null ? new Date(rs.getDate("LoginTime").getTime()) : null);
                driverlogininfo.setVehId(rs.getString("VehId"));


                sourceContext.collect(driverlogininfo);
            }


            Thread.sleep(30 * 60 * 1000);
        }
    }

    @Override
    public void cancel() {
        this.stop = true;
        try {

            if (connection != null) { //关闭连接和释放资源
                connection.close();
            }
            if (ps != null) {
                ps.close();
            }
        } catch (Exception e) {

        }

    }
}
